package com.example.rushbuy.consumer;

import com.example.rushbuy.LockObjHolder;
import com.example.rushbuy.event.BuyEvent;
import com.example.rushbuy.event.CheckTimeoutOrderEvent;
import com.example.rushbuy.mapper.OrderMapper;
import com.example.rushbuy.mapper.RushItemMapper;
import com.example.rushbuy.po.Order;
import com.example.rushbuy.po.RushItem;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Isolation;
import org.springframework.transaction.annotation.Transactional;

@Service
@RocketMQMessageListener(topic = "rushbuy", consumerGroup = "rushbuy-consumer-1")
public class MqConsumer implements RocketMQListener<BuyEvent> {

    @Autowired
    private OrderMapper orderMapper;
    @Autowired
    private RocketMQTemplate rocketMQTemplate;
    @Autowired
    private RushItemMapper rushItemMapper;

    @Override
    @Transactional(rollbackFor = Exception.class, isolation = Isolation.READ_COMMITTED)
    public void onMessage(BuyEvent buyEvent) {
//        System.out.println(buyEvent);
        RushItem rushItem = rushItemMapper.selectByPrimaryKey(buyEvent.getItemId());
        Integer test = rushItem.getStore();
//        System.out.println("用户ID：" + buyEvent.getUserId()+",当前库存：" + rushItem.getStore() + "，预设为：" + (rushItem.getStore() - buyEvent.getCount()));

        //生成订单
        Order order = new Order();
        order.setItemId(buyEvent.getItemId());
        order.setUserId(buyEvent.getUserId());
        order.setTotalPrice(buyEvent.getTotalPrice());
        order.setBuyCount(buyEvent.getCount());
        //设置为未支付状态
        order.setPayState(0);
        //设置为未超时状态
        order.setIsTimeout(false);
        orderMapper.insert(order);
        //减少库存
        //此处高并发下会有问题，需要加一个分布式锁
        //1.redis
        //2.zookeeper
        //3.database
        //4.模拟本地锁，依然会有问题，原因：事务没提交，锁先释放了
//        synchronized (LockObjHolder.lock) {
//        rushItem = rushItemMapper.selectByPrimaryKey(order.getItemId());
            rushItem = rushItemMapper.findById(order.getItemId());
        //不能使用事务隔离级别解决并发问题，因为并发时，java代码操作不是原子的
            if(!test.equals(rushItem.getStore())){
                System.out.println("!!!!!!出现不一致的情况！！！！第一次读取时库存为:"+test+",第二次读取库存为："+rushItem.getStore());
                System.out.println("用户ID：" + buyEvent.getUserId()+",当前库存：" + rushItem.getStore() + "，预设为：" + (rushItem.getStore() - order.getBuyCount()));
            }
        rushItem = rushItemMapper.selectByPrimaryKeyLock(order.getItemId());
        rushItem.setStore(rushItem.getStore() - order.getBuyCount());
        rushItemMapper.updateByPrimaryKey(rushItem);
        //插入成功后，发送一条延迟消息
        CheckTimeoutOrderEvent timeoutOrderEvent = new CheckTimeoutOrderEvent();
        timeoutOrderEvent.setOrder(order);
        Message message = new GenericMessage(timeoutOrderEvent);
        //延迟10s消费
        rocketMQTemplate.syncSend("rushbuy-delay", message, 1000l, 3);
//        }
    }
}
